package com.sdehualu.spark

import com.sdehualu.utils._
import net.minidev.json.JSONObject
import net.minidev.json.parser.JSONParser
import org.apache.commons.cli.{CommandLine, HelpFormatter, Option, Options, PosixParser}
import org.apache.spark.sql.types.{FloatType, StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SparkSession}

import scala.Array.newBuilder

/**
 * @Author: 吴敬超
 * @Date: 2021/7/21 9:47
 */
object SparkJsonUpHDFSNC2CSV {
  def main(args: Array[String]): Unit = {

    var cmd: CommandLine = null
    val options = new Options()
    try {
      //      hdfs url
      val HDFSurl = new Option("hdfsurl", true, "")
      HDFSurl.setRequired(true)
      options.addOption(HDFSurl)

      //      hdfs nc文件路径
      val HDFSfile = new Option("hdfsfile", true, "")
      HDFSfile.setRequired(true)
      options.addOption(HDFSfile)


      //      要转换的nc文件的变量名称
      val ncvarname = new Option("ncvarname", true, "")
      ncvarname.setRequired(true)
      options.addOption(ncvarname)


      //      输出到hdfs的路径
      val hdfsout = new Option("hdfsout", true, "")
      hdfsout.setRequired(true)
      options.addOption(hdfsout)


      //    其他环境上的文件
      val upfile = new Option("upfile", true, "")
      upfile.setRequired(true)
      options.addOption(upfile)


      val parser = new PosixParser()
      cmd = parser.parse(options, args)
    } catch {
      case e: Exception =>
        println("Submit SparkEngine Error!!!!!!!!!!!!!!!!!!")
        println(e.getMessage)
        println(e.printStackTrace())
        val formatter = new HelpFormatter()
        formatter.printHelp("This program args list:", options)
        System.exit(1)
    }
    val inputfileurl = cmd.getOptionValue("hdfsurl")
    println("inputfileurl is:" + inputfileurl)

    val inputfile = cmd.getOptionValue("hdfsfile")
    println("inputfile is:" + inputfile)

    val outputfile = cmd.getOptionValue("hdfsout")
    println("outputfile is:" + outputfile)

    val varname = cmd.getOptionValue("ncvarname")
    println("varname is:" + varname)
    val fileflag = new HDFSUtils()


    //********************************解析upfile json******************************************
    val strjson = cmd.getOptionValue("upfile")

    println("strjson is:")
    println(strjson)

    val jsonUtils = new JsonUtils()
    jsonUtils.jsonfile(strjson, inputfileurl, inputfile)


    //********************************解析upfile json******************************************


    //    如果输入文件或不存在那么不继续
    if (!fileflag.FileExist(inputfileurl, inputfile)) {

      println("输入文件不存在")

      System.exit(1)
    }



    //    如果需要输出的文件已经存在，不继续
    if (fileflag.FileExist(inputfileurl, outputfile)) {

      println("输出文件已存在")

      System.exit(1)
    }


    //    构建hdfs上nc文件的数据集
    val ncds = NetCDFUtils.loadDFSNetCDFDataSet(inputfileurl, inputfile, 4096, false)

    val nciu = new NetCDFInternalUtils()


    //    nc文件中所有var的集合
    val varlists = ncds.getVariables
    val varflg = nciu.Varflag(varlists, varname)
    //    如果文件中不存在变量，不继续
    if (!varflg) {
      println("此nc文件中不存在变量" + varname)
      System.exit(1)

    }

    //    得到要转换的数据
    //    val csvvar = ncds.findVariable(varname)

    //    转换的数据转成object 形式的array
    //    val csvvarread = csvvar.read().copyToNDJavaArray()

    //    转换变量的维度  [time = 24;, lat = 1101;, lon = 1051;]
    val varDimensionsSize = nciu.vards(ncds, varname)
    //    0维的数组不转
    //    var dimension 的个数
    val vardiss = varDimensionsSize.size()
    if (vardiss == 0) {
      println("nc文件中变量0维: " + varname)
      System.exit(1)

    }

    //    程序到这里说明至少存在一维数据，这里是第一纬度,1 to 1 也可以获取
    var listvars: List[Array[String]] = Nil
    for (i <- 0 until vardiss) {
      //    一维数据的结构[time = 24;]
      //    先判断第一纬度是否在var list 中
      //    第一个变量的名

      //     每一个名
      //      var vna = varDimensionsSize.get(i).getName

      //      如果var的dimension 在var的list中，从var 中读取
      if (nciu.Varflag(varlists, varDimensionsSize.get(i).getName)) {

        listvars = listvars :+ nciu.var1D(ncds.findVariable(varDimensionsSize.get(i).getName), "yyyy-MM-dd_HH:mm:ss")

        //        listvars = listvars :+ evarr
      } else {


        val b = newBuilder[String]
        val rangarrsize = varDimensionsSize.get(i).getLength
        b.sizeHint(rangarrsize)

        var rangi = 1
        while (i <= rangarrsize) {
          b += rangi.toString
          rangi += 1
        }
        listvars = listvars :+ b.result()
      }
    }


    var dataparlist: List[Int] = Nil

    //    第一个纬度的长度
    val firdsleng = varDimensionsSize.get(0).getLength

    val firdslengflag = firdsleng > 1
    //   如果变量第一个dimension 的size 大于1,根据第一个dimension 的size 分区

    //    如果第一维大小不大于1 ，并且存在第二位 用第二位
    if ((!firdslengflag) && (vardiss >= 2)) {

      dataparlist = List.range(0, listvars(1).length)

    } else {
      dataparlist = List.range(0, firdsleng)

    }


    println(dataparlist)
    /**
     * 转换条件满足，开始进行spark转换
     */

    val spark = SparkSession
      .builder()
      .appName("sparknc2csv")
      //            .master("local[6]")
      .getOrCreate()
    val sc = spark.sparkContext


    //    创建rdd
    //        val parRDD = sc.parallelize(List(1, 2, 3, 4, 5, 6), 6)

    //    val data = List(1, 2, 3, 4, 5, 6)
    val nameRDD = sc.parallelize(dataparlist, dataparlist.length)


    val rowrdd = nameRDD.map(everypar => {

      val ncf = NetCDFUtils.loadDFSNetCDFDataSet(inputfileurl,
        inputfile,
        4096, false)

      val csvvar = ncf.findVariable(varname)


      var listrow: List[Row] = Nil
      listrow = Var2Row.swithlist(everypar, vardiss, csvvar, listvars, firdslengflag)

      listrow


    }

    )


    //   把r4 的每一个压扁
    val csvr = rowrdd.flatMap(a => {

      a
    }

    )

    var liststructFields: List[StructField] = Nil

    for (strnum <- 0 to vardiss) {

      //      println(strnum)

      if (strnum == vardiss) {
        liststructFields = liststructFields :+ StructField(varname, FloatType, true)

      } else {

        liststructFields = liststructFields :+ StructField(varDimensionsSize.get(strnum).getName, StringType, true)

      }
      //      println(liststructFields)

    }

    //    创建schema
    val structType = StructType(liststructFields)

    //    通过rdd创建DataFrame
    val df = spark.createDataFrame(csvr, structType)
    //    把数据show出来
    //    df.show
    //    写成csv文件
    df.repartition(1).write
      .option("header", "true")
      //      .csv("./nccsvfile")

      //      存储到hdfs
      .csv(inputfileurl + outputfile)


    //停止sc，结束该任务
    sc.stop()

  }
}
